Question, example for RTP and replicate dvblast?

[SonnyWalkman]

Hi @ziutek ,

I'm using dvblast currently for streaming 4 HD channels filter a service from each MPTS each on different frequencies. I have quad dvb adaptor. Each TS stream is RTP with dvblastctl checking fe_stats and get_PMT every couple of seconds. I was wondering is your library is able to replicate filtering out one service and streaming over RTP? I'd like to keep all my code in go and not using a wrapper around dvblast for each service?

[SonnyWalkman]

An attempt to add RTP using UDP and PkWriter. Also send SDP announce packet for VLC to initiate reception. To be tested. Thought I'd lay it out for others assist.

package main

import (
    "fmt"
    "log"
    "net"
    "time"

    "github.com/ziutek/dvb"
    "github.com/ziutek/dvb/ts"
)

// outputUDP structure
type outputUDP struct {
    conn           *net.UDPConn
    addr           *net.UDPAddr
    buf            []byte
    sequenceNumber uint16
    timestamp      uint32
}

// WritePkt implements the ts.PktWriter interface
func (o *outputUDP) WritePkt(pkt ts.Pkt) error {
    rtpPacket := newRTPPacket(pkt[:], o.sequenceNumber, o.timestamp)
    o.sequenceNumber++
    o.timestamp += 3600 // Increment timestamp (adjust as needed)

    serializedPacket := rtpPacket.serialize()

    o.buf = append(o.buf, serializedPacket...)
    if len(o.buf) >= 7*188 { // Adjust buffer size if needed
        _, err := o.conn.WriteToUDP(o.buf, o.addr)
        if err != nil {
            return err
        }
        o.buf = o.buf[:0]
    }
    return nil
}

// newOutputUDP creates a new UDP output
func newOutputUDP(src, dst string) ts.PktWriter {
    var err error
    saddr := &net.UDPAddr{IP: net.IPv4zero, Port: 0}
    if src != "" {
        saddr, err = net.ResolveUDPAddr("udp", src)
        checkErr(err)
    }
    conn, err := net.ListenUDP("udp", saddr)
    checkErr(err)
    daddr, err := net.ResolveUDPAddr("udp", dst)
    checkErr(err)

    return &outputUDP{
        conn: conn,
        addr: daddr,
        buf:  make([]byte, 0, 7*188),
        sequenceNumber: 0,
        timestamp:      0,
    }
}

func checkErr(err error) {
    if err != nil {
        log.Fatalf("Error: %v", err)
    }
}

// RTP packet structure
type RTPPacket struct {
    Version        uint8
    Padding        uint8
    Extension      uint8
    CSRCCount      uint8
    Marker         uint8
    PayloadType    uint8
    SequenceNumber uint16
    Timestamp      uint32
    SSRC           uint32
    Payload        []byte
}

// RTP packetizer function
func newRTPPacket(payload []byte, sequenceNumber uint16, timestamp uint32) *RTPPacket {
    return &RTPPacket{
        Version:        2,
        PayloadType:    33, // Payload type for MPEG-TS
        SequenceNumber: sequenceNumber,
        Timestamp:      timestamp,
        SSRC:           1, // SSRC identifier
        Payload:        payload,
    }
}

// Serialize the RTP packet to bytes
func (pkt *RTPPacket) serialize() []byte {
    packet := make([]byte, 12+len(pkt.Payload))

    packet[0] = (pkt.Version << 6) | (pkt.Padding << 5) | (pkt.Extension << 4) | pkt.CSRCCount
    packet[1] = (pkt.Marker << 7) | pkt.PayloadType
    packet[2] = byte(pkt.SequenceNumber >> 8)
    packet[3] = byte(pkt.SequenceNumber & 0xFF)
    packet[4] = byte(pkt.Timestamp >> 24)
    packet[5] = byte(pkt.Timestamp >> 16)
    packet[6] = byte(pkt.Timestamp >> 8)
    packet[7] = byte(pkt.Timestamp & 0xFF)
    packet[8] = byte(pkt.SSRC >> 24)
    packet[9] = byte(pkt.SSRC >> 16)
    packet[10] = byte(pkt.SSRC >> 8)
    packet[11] = byte(pkt.SSRC & 0xFF)

    copy(packet[12:], pkt.Payload)

    return packet
}

// Generate SDP description
func generateSDP(dst string, port int) string {
    return fmt.Sprintf(`v=0
o=- 0 0 IN IP4 %s
s=No Name
c=IN IP4 %s
t=0 0
m=video %d RTP/AVP 33
a=rtpmap:33 MP2T/90000
`, dst, dst, port)
}

// Send SDP description
func sendSDP(sdp string, addr string) {
    conn, err := net.Dial("udp", addr)
    checkErr(err)
    defer conn.Close()

    _, err = conn.Write([]byte(sdp))
    checkErr(err)
}

func main() {
    // Open DVB device
    dev, err := dvb.Open("/dev/dvb/adapter0/dvr0")
    checkErr(err)
    defer dev.Close()

    // Create a UDP connection for RTP streaming
    rtpWriter := newOutputUDP("", "127.0.0.1:5004")

    // Generate and send SDP description
    sdp := generateSDP("127.0.0.1", 5004)
    sendSDP(sdp, "127.0.0.1:5005") // Assuming the receiver listens for SDP on port 5005

    // Wait for the SDP to be processed by the receiver
    time.Sleep(2 * time.Second)

    buf := make([]byte, 188) // MPEG-TS packet size

    for {
        _, err := dev.Read(buf)
        if err != nil {
            log.Fatalf("Error reading from DVB device: %v", err)
        }

        // Write the RTP packet using the ts.PktWriter interface
        rtpWriter.WritePkt(ts.Pkt(buf))
    }
}

[ziutek]

Interesting. I've never used RTP for my needs. Must definitely try your example. What does you use for receiving RTP? VLC?

[TVforME]

Hello @ziutek,

VLC over a congested network. VLC is very picky using raw UDP however, RTP divides both the video and audio streams into two separate streams and adds timestamps the the packets for reliability. SDP is an attempt to notify VLC of what is in the steams. Its a requirement to forward to GStreamer for another project on the go.

I've been using your ibrary and works great!

Question? How do you get the PIDS (videoPID, AudioPIDS and PCR_PID) knowing the PMT PID for the service? There isn't an example and I'm not sure what functions to use to do so?

I've managed to get a serviceID and PMT PID however, unable to get the group of PID's for the service using the pmt.Update(d) panic'ed..

Is it easy enough the add a function to return in a slice the PID's in a MPTS stream with > 1 service? Pass in int like 2 to return the PIDs sevice info etc?

Basically I'm receiving a FTA MPTS (multi program Transport Stream) which comes in via a couple of DVB USB sticks..

I'd like to get the first service available in the MPTS and stream the program as a single program source SPTS however to do this, I need to know all the PIDS associated with the service to pass into the demux filter.

It be benificial to have a function to simulate a set top box. Return the types of PES like HD, h264 video, AAC, ac3 audio teletext to etc typically like a DVB-T or DVB-S setop box does?

[ziutek]

I'd like to get the first service available in the MPTS and stream this a single SPTS however to do this, I need to know all the PIDS associated with the service. Be great to get the type of PES like HD, AAC, teletext etc typically like a DVB-T or DVB-S set top box would do?

Everything apart the PAT and SDT PIDs is dynamic in MPEG-TS. So to handle this properly you should track changes in PAT and PMT.

I developed a full fledged TS multiplexer using this library. But it's a closed source project.

Simplified solution for static, single program TS is to decode PAT table using the SectionDecoder and PAT.Update. Nex obtain from PAT the first program/service and determine its PMTPID. Next use another SectionDecoder to read PMT section and determine the PIDs of all streams.

In the real TS muxer all this is handled non-stop, concurently and the PAT decoder goroutnie communicates the changes in PAT to the PMT decoding goroutines using channels. They update filters used by stream readers, etc. It's quiet complicated dynamic system.

[SonnyWalkman]

Hi @ziutek,

I developed a full fledged TS multiplexer using this library. But it's a closed source project. Great, to use the library for creating a DVB-TS.. Looks to be all there for decode however but minimal for encoding? Am I mistaken?

Do you have a code snippet to return the PIDS associated with the PMT. I have managed to use parts of the examples to get PMT from the service ID.